package com.ionehe.group.common.notify.rocketmq.consumer;

import com.ionehe.group.common.notify.autoconfigure.AsyncVendors;
import com.ionehe.group.common.notify.configure.NotifyConfiguration;
import com.ionehe.group.common.notify.message.consumer.Consumer;
import com.ionehe.group.common.notify.message.exception.AsyncException;
import com.ionehe.group.common.notify.message.listener.MessageListener;
import com.ionehe.group.common.notify.rocketmq.flags.RocketKeyFlags;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Copyright (c) 2021 ionehe.com
 * Date: 2021/4/21
 * Time: 1:22 下午
 *
 * @author 2021年 <a href="mailto:a@ionehe.com">秀</a>
 * 【rocketmq消费者】
 */
public class ConsumerImpl implements Consumer {
    private final DefaultMQPushConsumer defaultMQPushConsumer;
    private final ConcurrentHashMap<String, MessageListener> subscribeTable = new ConcurrentHashMap<>();

    private final AtomicBoolean started = new AtomicBoolean(false);
    private final AtomicBoolean closed = new AtomicBoolean(false);


    public ConsumerImpl(NotifyConfiguration conf) {
        Map<String, Object> vendors = (Map<String, Object>) conf.getVendors().get(AsyncVendors.rocketmq);
        this.defaultMQPushConsumer = new DefaultMQPushConsumer((String) RocketKeyFlags.FLAG_CONSUMER_GROUP.get(vendors));
        this.defaultMQPushConsumer.setMessageModel(MessageModel.valueOf((String) RocketKeyFlags.FLAG_CONSUMER_MODEL.get(vendors)));
        this.defaultMQPushConsumer.setInstanceName((String) RocketKeyFlags.FLAG_INSTANCE_NAME.get(vendors));
        this.defaultMQPushConsumer.setNamesrvAddr((String) RocketKeyFlags.FLAG_NAMESRVADDR.get(vendors));
        int number = ((Integer) RocketKeyFlags.FLAG_CONSUMERTHREADNUMBER.get(vendors)).intValue();
        if (number > 0) {
            this.defaultMQPushConsumer.setConsumeThreadMin(number);
            this.defaultMQPushConsumer.setConsumeThreadMax(number);
        }
    }


    @Override
    public void start() {
        this.defaultMQPushConsumer.registerMessageListener(new MessageListenerImpl());
        try {
            if (this.started.compareAndSet(false, true)) {
                this.defaultMQPushConsumer.start();
            }
        } catch (Exception e) {
            throw new AsyncException(e.getMessage(), e);
        }
    }


    @Override
    public void shutdown() {
        if (this.closed.compareAndSet(false, true)) {
            this.defaultMQPushConsumer.shutdown();
        }
    }


    @Override
    public void subscribe(String topic, String subExpression, MessageListener listener) {
        if (null == topic) {
            throw new AsyncException("topic is null");
        }

        if (null == listener) {
            throw new AsyncException("listener is null");
        }

        try {
            this.subscribeTable.put(topic, listener);
            this.defaultMQPushConsumer.subscribe(topic, subExpression);
        } catch (MQClientException e) {
            throw new AsyncException("defaultMQPushConsumer subscribe exception", e);
        }
    }


    @Override
    public void unsubscribe(String topic) {
        if (null != topic) {
            this.defaultMQPushConsumer.unsubscribe(topic);
        }
    }

    @Override
    public void setConsumeFromWhere(ConsumeFromWhere consumeFromWhere) {
        this.defaultMQPushConsumer.setConsumeFromWhere(consumeFromWhere);
    }


    @Override
    public boolean isStarted() {
        return this.started.get();
    }


    @Override
    public boolean isClosed() {
        return this.closed.get();
    }


    class MessageListenerImpl implements MessageListenerConcurrently {
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgsRMQList, ConsumeConcurrentlyContext contextRMQ) {
            MessageExt message = msgsRMQList.get(0);
            MessageListener listener = (MessageListener) ConsumerImpl.this.subscribeTable.get(message.getTopic());
            if (null == listener) {
                throw new AsyncException("MessageListener is null");
            }
            return listener.consume(message);
        }
    }
}